package tableapi;

import bean.SensorReading;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.java.BatchTableEnvironment;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.*;
import org.apache.flink.table.types.DataType;
import org.apache.flink.types.Row;

/**
 * @Description: TODO QQ1667847363
 * @author: xiao kun tai
 * @date:2021/11/8 16:46
 */
public class Table2_CommonApi {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        /**
         * 基于老版本的planner的流处理
         */
        EnvironmentSettings oldStreamingSettings = EnvironmentSettings.newInstance()
                .useOldPlanner()
                .inStreamingMode()
                .build();

        StreamTableEnvironment oldStreamTableEnv = StreamTableEnvironment.create(env, oldStreamingSettings);

        /**
         * 基于老版本的planner的批处理
         */
        ExecutionEnvironment batchEnv = ExecutionEnvironment.getExecutionEnvironment();
        BatchTableEnvironment oldBatchTableEnv = BatchTableEnvironment.create(batchEnv);


        /**
         * 基于Blink的流处理
         */
        EnvironmentSettings blinkStreamSetting = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();

        StreamTableEnvironment blinkStreamTableEnv = StreamTableEnvironment.create(env, blinkStreamSetting);


        /**
         * 基于Blink的批处理
         */
        EnvironmentSettings blinkBatchSettings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inBatchMode()
                .build();

        TableEnvironment blinkBatchTableEnv = TableEnvironment.create(blinkBatchSettings);


        //表的创建，连接外部系统，读取数据
        //读取文件数据
        String filePath = "src/main/resources/sensor.txt";
        tableEnv.connect(new FileSystem().path(filePath))
                .withFormat(new Csv())
//                .withFormat(new Json())
                .withSchema(new Schema()
                        .field("id", DataTypes.STRING())
                        .field("timestamp", DataTypes.BIGINT())
                        .field("temperature", DataTypes.DOUBLE())
                )
                .createTemporaryTable("inputTable");

        /**
         * Table Api
         */
        Table inputTable = tableEnv.from("inputTable");

//        inputTable.printSchema();

//        tableEnv.toAppendStream(inputTable,Row.class).print();


        /**
         * 查询转换
         *
         */
        Table resultTable = inputTable.select("id,temperature")
                .filter("id==='sensor_6'");

        //聚合统计
        Table aggTable = inputTable.groupBy("id")
                .select("id,id.count as count,temperature.avg as avgTemp");

        //SQL
        tableEnv.sqlQuery("select id,temperature from inputTable where id ='sensor_6'");
        Table sqlAggTable = tableEnv
                .sqlQuery("select id,count(id) as cnt,avg(temperature) as avgTemp from inputTable group by id");


        tableEnv.toAppendStream(resultTable,Row.class).print("result");

        tableEnv.toRetractStream(aggTable,Row.class).print("agg");

        tableEnv.toRetractStream(sqlAggTable,Row.class).print("sqlagg");

        env.execute();

    }
}
